/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package cn.edu.thu.laud.utils;

import java.nio.ByteBuffer;
import java.nio.charset.CharacterCodingException;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.datanucleus.ClassLoaderResolver;
import org.datanucleus.ObjectManager;
import org.datanucleus.exceptions.NucleusDataStoreException;
import org.datanucleus.metadata.AbstractClassMetaData;
import org.datanucleus.metadata.AbstractMemberMetaData;
import org.datanucleus.metadata.ColumnMetaData;
import org.datanucleus.metadata.DiscriminatorMetaData;
import org.datanucleus.metadata.IndexMetaData;
import org.datanucleus.metadata.InheritanceMetaData;
import org.datanucleus.metadata.InheritanceStrategy;
import org.datanucleus.metadata.MetaDataManager;
import org.datanucleus.metadata.Relation;
import cn.edu.thu.laud.utils.serde.ConverterContext;

/**
 * Utility class to convert instance data to Cassandra columns and data types
 * 
 * TODO Cache the data returned here for better performance
 * 
 * @author zhuoan
 * 
 */
public class MetaDataUtils {

	public static final Charset UTF8 = Charset.forName("UTF-8");
	public static final String DEFAULT_DISCCOL_NAME = "class_type";
	//A null place holder for the cached values
	private static final String STRING_NULL = "\uffff\uffff";

	private static ConcurrentMap<String, String> classToCfNames = new ConcurrentHashMap<String, String>();

	private static ConcurrentMap<AbstractMemberMetaData, ByteBuffer> fieldToColumnNames = new ConcurrentHashMap<AbstractMemberMetaData, ByteBuffer>();

	private static ConcurrentMap<AbstractMemberMetaData, String> fieldToIndexNames = new ConcurrentHashMap<AbstractMemberMetaData, String>();

	private static ConcurrentMap<String, Set<ByteBuffer>> classToSubclasses = new ConcurrentHashMap<String, Set<ByteBuffer>>();


	/**
	 * Get the column metadata for the class and fieldname
	 * 
	 * @param metaData
	 * @param absoluteFieldNumber
	 * @return
	 */
	public static ByteBuffer getColumnName(AbstractClassMetaData metaData,
			int absoluteFieldNumber) {

		AbstractMemberMetaData memberMetaData = metaData
				.getMetaDataForManagedMemberAtAbsolutePosition(absoluteFieldNumber);

		ByteBuffer cached = fieldToColumnNames.get(memberMetaData);
        
		if (cached != null) {
			cached.rewind();
//			try {
//				System.out.println(metaData.getFullClassName()+":colName"+absoluteFieldNumber+"="+ByteBufferUtil.string(cached));
//			} catch (CharacterCodingException e) {
//				// TODO Auto-generated catch block
//				e.printStackTrace();
//			}
			return cached;
		}

		String name = null;
		// Try the first column if specified
		ColumnMetaData[] colmds = memberMetaData.getColumnMetaData();
		if (colmds != null && colmds.length > 0) {
			name = colmds[0].getName();
		} else {
			name = memberMetaData.getName();
		}

		cached = ByteBufferUtil.bytes(name);
				
		fieldToColumnNames.putIfAbsent(memberMetaData, cached);

		return cached;

	}

	/**
	 * Get the column metadata for the class and fieldname
	 * 
	 * @param metaData
	 * @param absoluteFieldNumber
	 * @return
	 */
	public static ByteBuffer getIdentityColumn(AbstractClassMetaData metaData) {

		int[] pks = metaData.getPKMemberPositions();

		if (pks.length != 1) {
			throw new NucleusDataStoreException(
					"Currently only single field identity objects are allowed");
		}

		AbstractMemberMetaData memberMetaData = metaData
				.getMetaDataForManagedMemberAtAbsolutePosition(pks[0]);

		// Try the first column if specified
		ColumnMetaData[] colmds = memberMetaData.getColumnMetaData();
		if (colmds != null && colmds.length > 0) {
			return ByteBufferUtil.bytes(colmds[0].getName());
		}

		// TODO should we allow defaults?
		return ByteBufferUtil.bytes(memberMetaData.getName());

	}

	/**
	 * Get the column name from the meta data, if it's not specified the default
	 * name of DEFAULT_DISCCOL_NAME is returned
	 * 
	 * @param metaData
	 * @return
	 */
	public static ByteBuffer getDiscriminatorColumnName(
			DiscriminatorMetaData metaData) {
		String name = metaData.getColumnName();

		if (name == null) {
			name = DEFAULT_DISCCOL_NAME;
		}

		return ByteBufferUtil.bytes(name);
	}

	/**
	 * Get the column name from the meta data, if it's not specified the default
	 * name of DEFAULT_DISCCOL_NAME is returned
	 * 
	 * @param acmd
	 * @return
	 */
	public static ByteBuffer getDiscriminatorColumnName(AbstractClassMetaData acmd) {
		
		DiscriminatorMetaData meta = acmd.getDiscriminatorMetaData();

		if (meta == null) {
			return null;
		}

		return getDiscriminatorColumnName(meta);

	}
	


	/**
	 * Get the name of the index. Will return null if no index is defined. If
	 * one is, it takes the name assigned by the user, otherwise it will create
	 * an index in the format of <TableName>_<FieldName>
	 * 
	 * @param metaData
	 * @return
	 */
	public static String getIndexName(AbstractMemberMetaData fieldMetaData) {

		// already indexed, return it
		String name = fieldToIndexNames.get(fieldMetaData);

		if (STRING_NULL == name) {
			return null;
		}

		IndexMetaData metaData = fieldMetaData.getIndexMetaData();

		// no index defined, set it to null and cache it
		if (metaData == null) {
			fieldToIndexNames.putIfAbsent(fieldMetaData, STRING_NULL);
			return null;

		}

		name = metaData.getName();

		if (name == null) {
			StringBuffer nameBuffer = new StringBuffer();

			nameBuffer.append(fieldMetaData.getName()).append("_index");
			name = nameBuffer.toString();
		}
		
		fieldToIndexNames.putIfAbsent(fieldMetaData, name);

		return name;

	}

	/**
	 * Get the name of the column family. Uses table name, if one doesn't exist,
	 * it uses the simple name of the class. If this class should never be
	 * persisted directly (subclass table persistence) null is returned
	 * 
	 * @param metaData
	 * @return
	 */
	public static String getColumnFamily(AbstractClassMetaData metaData) {

		String passedClassName = metaData.getFullClassName();
		String cfName = classToCfNames.get(passedClassName);

		if (cfName != null) {
			return cfName;
		}

		AbstractClassMetaData current = metaData;
		InheritanceMetaData inheritance = null;
		String tableName = null;

		while (current != null) {
			tableName = current.getTable();

			if (tableName != null) {
				cfName = tableName;
				break;
			}

			inheritance = metaData.getInheritanceMetaData();

			if (inheritance != null) {

				InheritanceStrategy strategy = inheritance.getStrategy();

				if (InheritanceStrategy.NEW_TABLE.equals(strategy)) {
					cfName = metaData.getTable();

					if (cfName == null) {
						cfName = metaData.getEntityName();

					}

					break;
				}

				// this class should never be persisted directly, return null
				else if (InheritanceStrategy.SUBCLASS_TABLE.equals(strategy)) {
					return null;
				}

			}

			current = (AbstractClassMetaData) current
					.getSuperAbstractClassMetaData();

		}

		classToCfNames.put(passedClassName, cfName);

		return cfName;

	}

	/**
	 * Get all discriminators as strings for this class and all possible
	 * subclasses. Will return at a minimum the discriminator for the passed
	 * class
	 * 
	 * @param className
	 * @param clr
	 * @param ec
	 * @return
	 */
	public static Set<ByteBuffer> getDescriminatorValues(String className,
			ClassLoaderResolver clr, ObjectManager om, ConverterContext converter) {

		Set<ByteBuffer> descriminators = classToSubclasses.get(className);

		if (descriminators != null) {
			return descriminators;
		}

		descriminators = new HashSet<ByteBuffer>();

		MetaDataManager mdm = om.getMetaDataManager();

		AbstractClassMetaData metaData = mdm
				.getMetaDataForClass(className, clr);

		DiscriminatorMetaData discriminator = metaData
				.getDiscriminatorMetaData();

		ByteBuffer value = converter.getBytes(discriminator.getValue());
		
		descriminators.add(value);

		String[] subClasses = mdm.getSubclassesForClass(className, true);

		if (subClasses != null) {

			for (String subclassName : subClasses) {
				metaData = mdm.getMetaDataForClass(subclassName, clr);

				
				value = converter.getBytes(metaData.getDiscriminatorMetaData().getValue());

				descriminators.add(value);
			}
		}

		classToSubclasses.putIfAbsent(className, descriminators);

		return descriminators;
	}
	


	/**
	 * Create a slice predicate with all mapped fetch column lists
	 * 
	 * @param metaData
	 * @param fieldNumbers
	 * @return
	 */
	public static SlicePredicate getFetchColumnList(
			AbstractClassMetaData metaData, int[] fieldNumbers) {

		ByteBuffer[] fieldNames = new ByteBuffer[fieldNumbers.length];

		for (int i = 0; i < fieldNumbers.length; i++) {
			fieldNames[i] = getColumnName(metaData, fieldNumbers[i]);		
//			try {
//				System.out.println(metaData.getFullClassName()+":colName"+i+"="+ByteBufferUtil.string(fieldNames[i]));
//			} catch (CharacterCodingException e) {
//				// TODO Auto-generated catch block
//				e.printStackTrace();
//			}
		}
		List<ByteBuffer> asList = new ArrayList<ByteBuffer>(32);
        for (ByteBuffer colName : fieldNames)
            asList.add(colName);
        SlicePredicate predicate = new SlicePredicate();
        predicate.setColumn_names(asList);
		return predicate;
	}

	/**
	 * Create a slice predicate that will retreive the discriminator column if
	 * one doesn't exist, null is returned
	 * 
	 * @param metaData
	 * @param fieldNumbers
	 * @return
	 */
	public static SlicePredicate getDescriminatorColumn(
			AbstractClassMetaData metaData) {

		DiscriminatorMetaData discriminatorMetaData = metaData
				.getDiscriminatorMetaData();

		if (discriminatorMetaData == null) {
			return null;
		}

		ByteBuffer columnName = getDiscriminatorColumnName(discriminatorMetaData);
		SlicePredicate predicate = new SlicePredicate();
        predicate.setColumn_names(Arrays.asList(new ByteBuffer[]{columnName}));
		return predicate;
	}
	

    public static boolean isRelationSingleValued(int type)
    {
        if (type == Relation.ONE_TO_ONE_UNI || type == Relation.ONE_TO_ONE_BI || type == Relation.MANY_TO_ONE_BI)
        {
            return true;
        }
        return false;
    }

    public static boolean isRelationMultiValued(int type)
    {
        if (type == Relation.ONE_TO_MANY_UNI || type == Relation.ONE_TO_MANY_BI || type == Relation.MANY_TO_MANY_BI)
        {
            return true;
        }
        return false;
    }


}
